package com.chenxu.gmall.realtime.app.func;

import com.chenxu.gmall.realtime.common.GmallConstant;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * Date: 2021/07/20
 * Desc: 实现自定义函数
 */
/*
商品主题统计完毕之后

product_stats_0709
	商品SPU				点击次数		下单次数		加购次数
	小米手机			  50				80           100


经过分词之后，得到的结果
	关键词				点击次数		下单次数		加购次数
	小米 				  50				80           100
	手机 				  50				80           100

目标是统计：关键词热度--->Keyword_Stats_0820
	关键词 				 操作次数		    来源
	小米 				  50				Click
	小米 				  80				Order
	小米  				  100  				Cart
	手机 				  50				Click
	手机 				  80				Order
	手机  				  100  				Cart

获取热词		搜索  10    点击 9   下单  8   加购  7
	关键词 				 操作次数			来源	       评分
	小米				  90				搜索          10
	小米 				  50				Click          9
	小米 				  80				Order	 8
	小米  				  100  				Cart	  7
	手机 				  50				Click
	手机 				  80				Order
	手机  				  100  				Cart

	sum
	关键词     90 * 10
			   50 * 9
			   80 * 8
			   100 * 6
 */
@FunctionHint(output = @DataTypeHint("ROW<ct BIGINT,source STRING>"))
public class KeywordProductC2RUDTF extends TableFunction<Row> {
    public void eval(Long clickCt, Long cartCt, Long orderCt) {

        if(clickCt>0L) {
            //如上所示，输出两列；一列是次数，一列是来源；
            Row rowClick = new Row(2);
            //赋值次数；
            rowClick.setField(0, clickCt);
            //赋值来源；
            rowClick.setField(1, GmallConstant.KEYWORD_CLICK);
            collect(rowClick);
        }
        if(cartCt>0L) {
            Row rowCart = new Row(2);
            rowCart.setField(0, cartCt);
            rowCart.setField(1, GmallConstant.KEYWORD_CART);
            collect(rowCart);
        }
        if(orderCt>0) {
            Row rowOrder = new Row(2);
            rowOrder.setField(0, orderCt);
            rowOrder.setField(1, GmallConstant.KEYWORD_ORDER);
            collect(rowOrder);
        }

    }
}

